package com.example.service;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.example.client.ClientRequest;
import com.example.client.Request;
import com.example.model.Product;
import com.example.model.Transaction;
import com.example.util.CryptoUtil;
import com.example.util.Util;

import it.unisa.dia.gas.jpbc.Element;

@Service
public class ClientService {
    // private final static String ip = "172.22.132.90";
    private final static String ip = "127.0.0.1";

    private final static int[] arr = new int[] { 7601, 7602, 7603, 7604 };

    private final static String org1_private_key_path = "src\\main\\resources\\org1\\org1_sk.key";

    private final static String org1_public_key_path = "src\\main\\resources\\org1\\org1_pk.key";

    private static final Logger logger = LoggerFactory.getLogger(ClientService.class);

    // 新增统计变量
    private final ConcurrentHashMap<String, Long> requestStartTimes = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, Long> requestEndTimes = new ConcurrentHashMap<>();
    private final AtomicInteger totalRequests = new AtomicInteger(0);
    private final AtomicLong totalLatency = new AtomicLong(0);
    private volatile long testStartTime;
    private volatile long testEndTime;

    // 新增成员变量
    private final ConcurrentHashMap<String, Long> pendingRequests = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    private static final long REQUEST_TIMEOUT_MS = 20000; // 20秒超时

    // 初始化定时任务
    @PostConstruct
    public void init() {
        scheduler.scheduleAtFixedRate(this::checkTimeouts, 0, 10, TimeUnit.SECONDS);
    }

    private ConcurrentHashMap<String, String> requestToUserMap = new ConcurrentHashMap<>();

    public void associateRequestWithUser(String requestId, String username) {
        requestToUserMap.put(requestId, username);
    }

    private void checkTimeouts() {
        long now = System.currentTimeMillis();
        pendingRequests.entrySet().removeIf(entry -> {
            if (now - entry.getValue() > REQUEST_TIMEOUT_MS) {
                logger.warn("请求超时{}", entry.getKey());
                // 这里添加重试逻辑
                // retryRequest(entry.getKey());
                return true;
            }
            return false;
        });
    }

    public void onRequest(ClientRequest clientRequest) {
        String requestId = clientRequest.getRequest().getRequestId();
        // 记录请求开始时间
        long startTime = System.currentTimeMillis();
        requestStartTimes.put(requestId, startTime);
        // 记录请求开始时间
        pendingRequests.put(clientRequest.getRequest().getRequestId(), System.currentTimeMillis());
        for (int i = 0; i < arr.length; i++) {
            try {
                SocketChannel channel = SocketChannel.open();
                // channel.connect(new InetSocketAddress(ip, 7604));
                channel.socket().connect(new InetSocketAddress(ip, arr[i]), 3000);
                logger.info("发送消息到{}",arr[i]);
                sendMessage(channel, JSON.toJSONString(clientRequest));
                totalRequests.incrementAndGet(); // 总请求数+1
                return;
            } catch (Exception e) {
                // System.out.println("发送消息失败" + e.getMessage());
                logger.info("发送失败{}",e.getMessage());
                requestStartTimes.remove(requestId);
                pendingRequests.remove(requestId);
            }
        }
    }

    // ClientService.java 修改completeRequest方法
    public void completeRequest(String requestId) {

        if (pendingRequests.containsKey(requestId)) {
            long endTime = System.currentTimeMillis();
            Long startTime = requestStartTimes.get(requestId);
            if (startTime != null) {
                long latency = endTime - startTime;
                requestEndTimes.put(requestId, latency);
                totalLatency.addAndGet(latency); // 累加总时延
            }
            pendingRequests.remove(requestId);
            requestStartTimes.remove(requestId);
            logger.info("共识达成，取消计时{}", requestId);
            // 更新测试结束时间为最后一个请求完成的时间
            testEndTime = System.currentTimeMillis();
            printTestResults();
        }
    }

    // 发送消息到指定连接
    private static boolean sendMessage(SocketChannel channel, String msg) {
        try {
            ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
            // logger.info("发送消息 " + msg);
            channel.write(buffer);
            return true;
        } catch (IOException e) {
            System.err.println("发送消息失败: " + e.getMessage());
            close(channel);
            return false;
        }
    }

    // 关闭指定连接
    private static void close(SocketChannel channel) {
        try {
            channel.close();
        } catch (Exception e) {
            System.out.println("关闭指定连接 " + e.getMessage());
        }
    }

    public void sendConcurrentRequests(int numRequests, int threads)
            throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(threads);
        Semaphore rateLimiter = new Semaphore(1); // 每秒允许200个请求（5ms间隔）

        startTest();

        for (int i = 0; i < numRequests; i++) {
            rateLimiter.acquire(); // 获取许可
            executor.submit(() -> {
                try {
                    ClientRequest clientRequest = new ClientRequest();
                    Request request = new Request("client-" + UUID.randomUUID(), 0);
                    clientRequest.setRequest(request);
                    clientRequest.setClient("org1");
                    Element pk = Util.readPublicKey(org1_public_key_path);
                    Element sk = Util.readPrivateKey(org1_private_key_path);
                    clientRequest.sign(sk);
                    clientRequest.setPublicKey(Util.PkToPem(pk));
                    clientRequest.setDigest(CryptoUtil.SHA256(JSON.toJSONString(request)));
                    onRequest(clientRequest);
                } finally {
                    rateLimiter.release();
                }
            });
        }

        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.HOURS);
        // 结束测试
    }

    public void printTestResults() {
        System.out.println("====== PBFT性能测试结果 ======");
        System.out.printf("总请求数: %d\n", totalRequests.get());
        System.out.printf("完成请求数: %d\n", requestEndTimes.size());
        System.out.printf("平均吞吐量: %.2f 请求/秒\n", getThroughput());
        System.out.printf("平均时延: %.2f 毫秒\n", getAverageLatency());
        System.out.println("============================");
    }

    public double getThroughput() {
        long durationMs = testEndTime - testStartTime;
        if (durationMs == 0)
            return 0;
        return (totalRequests.get() * 1000.0) / durationMs; // 请求数/秒
    }

    public double getAverageLatency() {
        int completed = requestEndTimes.size();
        if (completed == 0)
            return 0;
        return totalLatency.get() / (double) completed; // 平均时延（毫秒）
    }

    public void startTest() {
        testStartTime = System.currentTimeMillis();
    }

    public void endTest() {
        testEndTime = System.currentTimeMillis();
    }

    public static void main(String[] args) {
        Product product = new Product();

        String owner = "e5e3e72f0c186eb217ab9aae7d84122d86e4cfc4c61367d48c70d7b3cc74242b";
        product.setOwner(owner);// owner
        product.setOrg("org1");
        product.setAmount(100);
        product.setPrice(0.56);
        product.setType("wind");

        Element pk = Util.readPublicKey(org1_public_key_path);
        Element sk = Util.readPrivateKey(org1_private_key_path);
        String clientId = CryptoUtil.SHA256(Util.PkToPem(pk));
        Request request = new Request(clientId, 0);// 客户端标识
        product.setId(request.getRequestId());
        String productJson = JSON.toJSONString(product);
        request.setData(productJson); // 扩展data字段

        ClientRequest clientRequest = new ClientRequest();
        clientRequest.setRequest(request);
        clientRequest.setClient("org1");
        clientRequest.sign(sk);
        clientRequest.setPublicKey(Util.PkToPem(pk));
        clientRequest.setDigest(CryptoUtil.SHA256(JSON.toJSONString(request)));
        try {
            SocketChannel channel = SocketChannel.open();
            channel.connect(new InetSocketAddress(ip, 7603));
            sendMessage(channel, JSON.toJSONString(clientRequest));
        } catch (Exception e) {
            System.out.println("发送消息失败" + e.getMessage());
        }

        try {
            Thread.sleep(5000);
        } catch (Exception e) {
            logger.warn("发送失败{}", e.getMessage());
        }

        Transaction transaction = new Transaction();
        transaction.setBuyer(owner);
        transaction.setAmount(100);
        transaction.setProductId(product.getId());
        transaction.setSaler(owner);
        transaction.setBalance((float) 56.0);
        Request orderRequest = new Request(clientId, 1);

        transaction.setRequestId(orderRequest.getRequestId());
        String orderJson = JSON.toJSONString(transaction);
        orderRequest.setData(orderJson);

        ClientRequest orderClientRequest = new ClientRequest();
        orderClientRequest.setRequest(orderRequest);
        orderClientRequest.setClient("org1");
        orderClientRequest.sign(sk);
        orderClientRequest.setPublicKey(Util.PkToPem(pk));
        orderClientRequest.setDigest(CryptoUtil.SHA256(JSON.toJSONString(orderRequest)));
        try {
            SocketChannel channel = SocketChannel.open();
            channel.connect(new InetSocketAddress(ip, 7603));
            sendMessage(channel, JSON.toJSONString(orderClientRequest));
        } catch (Exception e) {
            System.out.println("发送消息失败" + e.getMessage());
        }
    }
}
